package cn.smp.client;

import cn.smp.client.handler.RpcResponseMessageHandler;
import cn.smp.message.RpcRequestMessage;
import cn.smp.protocol.MessageCodecSharable;
import cn.smp.protocol.ProcaotolFrameDecoder;
import cn.smp.protocol.SequenceIdGenerator;
import cn.smp.server.service.HelloService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInboundHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOutboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Proxy;

/**
 * 客户端
 * 写消息->出站顺序处理
 * @see ChannelOutboundHandler implements
 *
 * 读消息—>入站顺序处理
 * @see ChannelInboundHandler implements
 */
@Slf4j
public class RpcClientManager {

    private static volatile Channel channel = null;
    private static final Object LOCK = new Object();

    public static void main(String[] args) {
//        getChannel().writeAndFlush(new
//                RpcRequestMessage(
//                1,
//                "cn.smp.server.service.HelloService",
//                "sayHello",
//                String.class,
//                new Class[]{String.class},
//                new Object[]{"张三"}));
        final HelloService service = getProxyService(HelloService.class);
        System.out.println(service.sayHello("张三"));
//        System.out.println(service.sayHello("李四"));
//        System.out.println(service.sayHello("王五"));
    }

    //创建代理类
    public static <T> T getProxyService(Class<T> serviceClass){
        ClassLoader classLoader = serviceClass.getClassLoader();
        Class<?>[] classes = new Class[]{serviceClass};
        final Object o = Proxy.newProxyInstance(classLoader, classes, (proxy, method, args) -> {
            //1.将方法的调用转换为 消息对象
            final int sequenceId = SequenceIdGenerator.nextId();
            final RpcRequestMessage msg = new RpcRequestMessage(
                    sequenceId,
                    serviceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args);

            //2.将消息对象发送出去
            getChannel().writeAndFlush(msg);

            //3.准备一个空promise对象，来接收结果                            指定promise对象异步接收结果线程
            final DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
            RpcResponseMessageHandler.PROMISES.put(sequenceId,promise);

//            promise.addListener(future -> {
//                //线程
//            });

            //4.等待promise 结果
            promise.await();
            if (promise.isSuccess()) {
                //调用正常
                return promise.getNow();
            } else {
                //调用异常
                throw new RuntimeException(promise.cause());
            }
        });
        return (T) o;
    }

    /**
     * 获取唯一的channle对象
     * @return
     */
    public static Channel getChannel(){
        if (channel != null) {
            return channel;
        }
        synchronized (LOCK){
            if (channel != null) {
                return channel;
            }
            initChannel();
            return channel;
        }
    }

    private static void initChannel() {
        final LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        final MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        //rpc响应消息处理
        final RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();

        final NioEventLoopGroup worker = new NioEventLoopGroup();
        final Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(worker);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProcaotolFrameDecoder());
                ch.pipeline().addLast(LOGGING_HANDLER);
                ch.pipeline().addLast(MESSAGE_CODEC);
                ch.pipeline().addLast(RPC_HANDLER);
            }
        });
        try {
            channel = bootstrap.connect("localhost", 8080).sync().channel();
            channel.closeFuture().addListener(future -> {
                worker.shutdownGracefully();
            });
        } catch (InterruptedException e) {
            log.error("client error",e);
        }
    }
}
